package com.kwdb.server.service;

import com.kwdb.grpc.*;
import com.kwdb.server.storage.StorageEngine;
import com.kwdb.server.query.QueryEngine;
import com.kwdb.server.parser.LineProtocolParser;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * KWDB gRPC服务实现
 */
public class KWDBServiceImpl extends KWDBServiceGrpc.KWDBServiceImplBase {
    
    private static final Logger logger = LoggerFactory.getLogger(KWDBServiceImpl.class);
    
    private final StorageEngine storageEngine;
    private final QueryEngine queryEngine;
    private final LineProtocolParser parser;
    
    // 统计信息
    private final AtomicLong totalRequests = new AtomicLong(0);
    private final AtomicLong totalWrites = new AtomicLong(0);
    private final AtomicLong totalQueries = new AtomicLong(0);
    private final AtomicLong activeConnections = new AtomicLong(0);
    
    public KWDBServiceImpl(StorageEngine storageEngine) {
        this.storageEngine = storageEngine;
        this.queryEngine = new QueryEngine(storageEngine);
        this.parser = new LineProtocolParser();
    }
    
    @Override
    public StreamObserver<WriteRequest> writeStream(StreamObserver<WriteResponse> responseObserver) {
        activeConnections.incrementAndGet();
        totalRequests.incrementAndGet();
        
        StreamObserver<WriteRequest> requestObserver = new StreamObserver<WriteRequest>() {
            private long totalPoints = 0;
            
            @Override
            public void onNext(WriteRequest request) {
                try {
                    // 处理写入请求
                    List<DataPoint> points = request.getPointsList();
                    totalPoints += points.size();
                    
                    // 批量写入到存储引擎
                    storageEngine.writeBatch(points);
                    totalWrites.addAndGet(points.size());
                    
                    logger.debug("流式写入处理完成，数据点数量: {}", points.size());
                    
                } catch (Exception e) {
                    logger.error("流式写入处理失败", e);
                    responseObserver.onError(io.grpc.Status.INTERNAL
                            .withDescription("流式写入失败: " + e.getMessage())
                            .asRuntimeException());
                }
            }
            
            @Override
            public void onError(Throwable t) {
                logger.error("流式写入客户端错误", t);
                activeConnections.decrementAndGet();
            }
            
            @Override
            public void onCompleted() {
                // 发送成功响应
                WriteResponse response = WriteResponse.newBuilder()
                        .setSuccess(true)
                        .setMessage("流式写入完成")
                        .setWrittenCount((int) totalPoints)
                        .setExecutionTimeMs(System.currentTimeMillis())
                        .build();
                
                responseObserver.onNext(response);
                responseObserver.onCompleted();
                
                activeConnections.decrementAndGet();
                logger.info("流式写入完成，总数据点: {}", totalPoints);
            }
        };
        
        // 返回请求观察者
        return requestObserver;
    }
    
    @Override
    public void writeBatch(WriteBatchRequest request, StreamObserver<WriteResponse> responseObserver) {
        activeConnections.incrementAndGet();
        totalRequests.incrementAndGet();
        
        long startTime = System.currentTimeMillis();
        
        try {
            List<DataPoint> points = request.getPointsList();
            
            // 批量写入到存储引擎
            storageEngine.writeBatch(points);
            totalWrites.addAndGet(points.size());
            
            long executionTime = System.currentTimeMillis() - startTime;
            
            WriteResponse response = WriteResponse.newBuilder()
                    .setSuccess(true)
                    .setMessage("批量写入成功")
                    .setWrittenCount(points.size())
                    .setExecutionTimeMs(executionTime)
                    .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            
            logger.info("批量写入完成，数据点数量: {}, 执行时间: {}ms", points.size(), executionTime);
            
        } catch (Exception e) {
            logger.error("批量写入失败", e);
            
            WriteResponse response = WriteResponse.newBuilder()
                    .setSuccess(false)
                    .setMessage("批量写入失败: " + e.getMessage())
                    .setErrorCode("WRITE_ERROR")
                    .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } finally {
            activeConnections.decrementAndGet();
        }
    }
    
    @Override
    public void write(WriteRequest request, StreamObserver<WriteResponse> responseObserver) {
        activeConnections.incrementAndGet();
        totalRequests.incrementAndGet();
        
        long startTime = System.currentTimeMillis();
        
        try {
            List<DataPoint> points = request.getPointsList();
            
            // 写入到存储引擎
            for (DataPoint point : points) {
                storageEngine.write(point);
            }
            totalWrites.addAndGet(points.size());
            
            long executionTime = System.currentTimeMillis() - startTime;
            
            WriteResponse response = WriteResponse.newBuilder()
                    .setSuccess(true)
                    .setMessage("写入成功")
                    .setWrittenCount(points.size())
                    .setExecutionTimeMs(executionTime)
                    .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            
            logger.debug("写入完成，数据点数量: {}, 执行时间: {}ms", points.size(), executionTime);
            
        } catch (Exception e) {
            logger.error("写入失败", e);
            
            WriteResponse response = WriteResponse.newBuilder()
                    .setSuccess(false)
                    .setMessage("写入失败: " + e.getMessage())
                    .setErrorCode("WRITE_ERROR")
                    .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } finally {
            activeConnections.decrementAndGet();
        }
    }
    
    @Override
    public void query(QueryRequest request, StreamObserver<QueryResponse> responseObserver) {
        activeConnections.incrementAndGet();
        totalRequests.incrementAndGet();
        totalQueries.incrementAndGet();
        
        long startTime = System.currentTimeMillis();
        
        try {
            String sql = request.getSql();
            
            // 执行查询
            QueryEngine.QueryResult result = queryEngine.execute(sql);
            
            long executionTime = System.currentTimeMillis() - startTime;
            
            QueryResponse.Builder responseBuilder = QueryResponse.newBuilder()
                    .setSuccess(true)
                    .setMessage("查询成功")
                    .setExecutionTimeMs(executionTime);
            
            if (result.getColumns() != null) {
                responseBuilder.addAllColumns(result.getColumns());
            }
            if (result.getRows() != null) {
                responseBuilder.addAllRows(result.getRows());
            }
            responseBuilder.setTotalCount(result.getTotalCount())
                          .setHasMore(result.hasMore());
            
            QueryResponse response = responseBuilder.build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            
            int rowCount = (result.getRows() != null) ? result.getRows().size() : 0;
            logger.info("查询完成，SQL: {}, 结果行数: {}, 执行时间: {}ms", 
                       sql, rowCount, executionTime);
            
        } catch (Exception e) {
            logger.error("查询失败", e);
            
            QueryResponse response = QueryResponse.newBuilder()
                    .setSuccess(false)
                    .setMessage("查询失败: " + e.getMessage())
                    .setErrorCode("QUERY_ERROR")
                    .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } finally {
            activeConnections.decrementAndGet();
        }
    }
    
    @Override
    public void queryStream(QueryRequest request, StreamObserver<QueryResponse> responseObserver) {
        activeConnections.incrementAndGet();
        totalRequests.incrementAndGet();
        totalQueries.incrementAndGet();
        
        try {
            String sql = request.getSql();
            
            // 执行流式查询（简化实现：直接返回全部结果）
            QueryEngine.QueryResult result = queryEngine.execute(sql);
            
            // 分批返回结果
            if (result.getRows() != null) {
                for (com.kwdb.grpc.QueryRow row : result.getRows()) {
                    QueryResponse response = QueryResponse.newBuilder()
                            .setSuccess(true)
                            .setMessage("查询数据")
                            .addRows(row)
                            .build();
                    
                    responseObserver.onNext(response);
                }
            }
            
            responseObserver.onCompleted();
            activeConnections.decrementAndGet();
            logger.info("流式查询完成，SQL: {}", sql);
            
        } catch (Exception e) {
            logger.error("流式查询失败", e);
            
            QueryResponse response = QueryResponse.newBuilder()
                    .setSuccess(false)
                    .setMessage("流式查询失败: " + e.getMessage())
                    .setErrorCode("QUERY_ERROR")
                    .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
            activeConnections.decrementAndGet();
        }
    }
    
    @Override
    public void healthCheck(HealthRequest request, StreamObserver<HealthResponse> responseObserver) {
        HealthResponse response = HealthResponse.newBuilder()
                .setStatus(HealthResponse.ServingStatus.SERVING)
                .setMessage("KWDB服务正常运行")
                .setTimestamp(System.currentTimeMillis())
                .build();
        
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
    
    @Override
    public void getStats(StatsRequest request, StreamObserver<StatsResponse> responseObserver) {
        StatsResponse response = StatsResponse.newBuilder()
                .setTotalRequests(totalRequests.get())
                .setTotalWrites(totalWrites.get())
                .setTotalQueries(totalQueries.get())
                .setActiveConnections(activeConnections.get())
                .setAvgResponseTimeMs(0.0) // 需要实现响应时间统计
                .setMemoryUsageBytes(Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory())
                .build();
        
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
    
    /**
     * 转换查询结果为gRPC行
     * @param row 查询结果行
     * @return gRPC行
     */
    // 辅助方法已废弃，QueryEngine直接返回gRPC QueryRow对象
}
